package com.caul.demo.hadoop.mapreduce.join;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.net.URI;

/**
 * 在MapTask阶段完成, 不需要ReduceTask
 * Created by BlueDream on 2018-04-08.
 */
public class JoinCacheSubmitter {

    public static void main(String[] args)
            throws IOException, ClassNotFoundException, InterruptedException {

        int reducerTask = 1;
        if (args[0] != null) {
            reducerTask = Integer.parseInt(args[0]);
        }

        Path destPath = new Path(args[1]);


        Path[] srcPaths = new Path[args.length - 2];
        for (int i = 2; i < args.length; i++) {
            srcPaths[i - 2] = new Path(args[i]);
        }

        String fileCache = "profile.txt";
        String filePath = "hdfs://cdh-namenode:9000/join/" + fileCache;

        //本地执行
        fileCache = "file:/D:/workspace/ws-idea2016/bigdata/data/join/" + fileCache;
        filePath = fileCache;

        Configuration conf = new Configuration();
        conf.set("file.cache", fileCache);
        //本地的运行
        //    conf.set("mapreduce.framwork.name", "local");//mapred-default.xml
        //    conf.set("ds.defaultFS", "hdfs://cdh-namenode:9000/");//core-default.xml

        Job job = Job.getInstance(conf, "joincachedemo");

        //指定执行的jar包
        job.setJarByClass(JoinCacheSubmitter.class);
        //指定mapper信息
        job.setMapperClass(JoinCacheMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(JoinBean.class);
        //reduce设置0
        job.setNumReduceTasks(0);
        //告诉yarn源文件的读取类型
        job.setInputFormatClass(TextInputFormat.class);
        //设置读取的文件路径
        FileInputFormat.setInputPaths(job, srcPaths);
//        FileInputFormat.setInputPathFilter(job, JoinCachePathFilter.class);
        //设置输出结果
        job.setOutputFormatClass(TextOutputFormat.class);
        FileOutputFormat.setOutputPath(job, destPath);
        //设置reducer分区机制
        job.setNumReduceTasks(reducerTask);
        //将URI的文件下载到所有执行的机器上
        job.addCacheFile(URI.create(filePath));

        boolean flag = job.waitForCompletion(true);
        //客户端程序退出
        System.exit(flag ? 0 : 1);
    }
}
